package com.ibm.cps.message;

import java.util.Arrays;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.gson.Gson;

/**
 * 
 * @author Yaoliang
 *
 */
public class ParsedDataMessage extends DataMessage implements KryoSerializable{

	protected Long timestamp;
	protected Object[] values;
	protected Comparable tsKey;

	static private Logger logger = LoggerFactory.getLogger(ParsedDataMessage.class);

	public ParsedDataMessage(int len) {
		values = new Object[len];
		tsKey = null;
		timestamp = null;
	}
	
	public ParsedDataMessage(int len,Comparable tsKey, Long timestamp){
		values = new Object[len];
		this.tsKey = tsKey;
		this.timestamp = timestamp;
	}

	/*
	 * public ParsedDataMessage(long tsKey, long timestamp, Comparable[] values)
	 * { this.tsKey = tsKey; this.timestamp = timestamp; this.values = values;
	 * 
	 * // TODO messageBytes = new byte[0]; }
	 */

	@Override
	public Comparable getTsKey() {
		return tsKey;
	}

	public Long getTimestamp() {
		return timestamp;
	}

	public void setTsKey(Comparable value) {
		tsKey = value;
	}

	public void setTimestamp(long value) {
		timestamp = value;
	}

	@Deprecated
	public Comparable[] getValues() {
		Comparable[] _values = new Comparable[values.length];
		for (int i = 0; i < _values.length; i++) {
			Object value = values[i];
			
			if (value == null || !value.getClass().isArray()) {
				_values[i] = (Comparable) values[i];
			}else {
				double[] value_array = (double[]) value;
				String value_str = Arrays.toString(value_array);
				_values[i] = value_str;
			}
		}
		return _values;
//		return (Comparable[])values;
	}
	
	public Object[] getObjectValues() {
		return values;
	}
	

	/*
	 * public String getJsonMessage() { ObjectNode node =
	 * JSONUtility.newObjectNode(); node.put("k", tsKey); node.put("t",
	 * timestamp); node.put("v", Arrays.toString(values)); return
	 * node.toString(); }
	 */

	/*
	 * public String getReturnedMessage(JsonSchema schema) {
	 * schema.restoreFromDataMessage(this); Object[] tAndValues = new
	 * Object[values.length + 1]; tAndValues[0] = this.timestamp; for (int i =
	 * 0; i < values.length; i++) { tAndValues[i + 1] = values[i]; } ObjectNode
	 * node = reader.getJSONRecord(values, Constants.DATA_FLAG_SPARSE);
	 * node.put("k", this.tsKey); return node.toString(); }
	 */

	// Added for debug
	@Override
	public String toString() {
		Gson gson = new Gson();
		return gson.toJson(this);
	}

	/**
	 * private long timestamp; private Comparable[] values; private byte[]
	 * messageBytes;
	 */
	@Override
	public void write(Kryo kryo, Output output) {
		kryo.writeObject(output, tsKey);
//		output.writeLong(tsKey);
		output.writeLong(timestamp);
		kryo.writeObject(output, values);
	}

	@Override
	public void read(Kryo kryo, Input input) {
		tsKey = kryo.readObject(input, Comparable.class);
//		tsKey = input.readLong();
		timestamp = input.readLong();
		values = kryo.readObject(input, Comparable[].class);
	}

	
	public void setValue(int number, Comparable value) {
		if (number >= 0 && number < values.length) {
			values[number] = value;
		}else {
			logger.warn("tag number ({}) is out of range, skip the setValue.", number);
		}
	}

	public void setValue(int number, Object value) {
		if (number >= 0 && number < values.length) {
			values[number] = value;
		} else {
			logger.warn("tag number ({}) is out of range, skip the setValue.", number);
		}
	}
	
	// if you sure values[number] is not a array type,you can use this function
	public Comparable getValue(int number) {
		if (number >= 0 && number < values.length) {
			return (Comparable)values[number];
		}
		return null;
	}
	
	public Object[] getArrayValue(int number) {
		if (number >= 0 && number < values.length) {
			return (Object[]) values[number];
		}
		return null;
	}
	
	public <T> T getArray(int number){
		if (number >= 0 && number < values.length) {
			return (T) values[number];
		}
		return null;
	}

	public int getValueCount() {
		return values.length;
	}

}
